package com.wispx.gateway.register.service.impl.nacos;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingMaintainFactory;
import com.alibaba.nacos.api.naming.NamingMaintainService;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.Service;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import com.alibaba.nacos.shaded.io.grpc.netty.shaded.io.netty.util.concurrent.DefaultThreadFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wispx.gateway.config.config.Config;
import com.wispx.gateway.config.config.RegisterCenter;
import com.wispx.gateway.config.pojo.ServiceDefinition;
import com.wispx.gateway.config.pojo.ServiceInstance;
import com.wispx.gateway.register.service.RegisterCenterListener;
import com.wispx.gateway.register.service.RegisterCenterProcessor;
import com.wispx.gateway.register.util.NetUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class NacosRegisterCenter implements RegisterCenterProcessor {

    /**
     * 注册中心配置
     */
    private Config config;

    /**
     * 维护服务实例信息
     */
    private NamingService namingService;

    /**
     * 维护服务定义信息
     */
    private NamingMaintainService namingMaintainService;

    /**
     * 监听器
     */
    private RegisterCenterListener listener;

    /**
     * 是否初始化
     */
    private AtomicBoolean init = new AtomicBoolean(false);

    @SneakyThrows(Exception.class)
    @Override
    public void init(Config config) {
        if (!init.compareAndSet(false, true)) {
            return;
        }
        this.config = config;
        Properties properties = buildProperty(config.getRegisterCenter());
        // log.error(String.valueOf(properties));
        this.namingService = NamingFactory.createNamingService(properties);
        this.namingMaintainService = NamingMaintainFactory.createMaintainService(properties);
        String group = config.getRegisterCenter().getNacos().getGroup();

        // 查阅Nacos官方文档
        // 将网关注册到Nacos
        Instance instance = new Instance();
        instance.setIp(NetUtil.getLocalIp());
        instance.setPort(config.getPort());
        instance.setInstanceId(NetUtil.getLocalIp() + ":" + config.getPort());
        namingService.registerInstance(config.getName(), group, instance);
        log.info("success register gateway-instance {}", instance);

        // 网关元信息注册到Nacos
        Map<String, String> serviceMetaDateInfo = BeanUtils.describe(new ServiceDefinition(config.getName()));
        namingMaintainService.updateService(config.getName(), group, 0, serviceMetaDateInfo);
        log.info("success register gateway-service-meta  {}", serviceMetaDateInfo);
    }

    // 起一个定时任务不断地订阅所有服务（namingService.subscribe(serverName, group, listener)）
    @Override
    public void subscribeServiceChange(RegisterCenterListener listener) {
        if (!init.get()) {
            return;
        }
        this.listener = listener;
        Executors.newScheduledThreadPool(1, new DefaultThreadFactory("subscribeAllServices"))
                .scheduleWithFixedDelay(() -> subscribeAllServices(), 0, 10, TimeUnit.SECONDS);
    }


    // 订阅Nacos中所有服务(只负责给服务实例添加监听器)
    public void subscribeAllServices() {

        try {
            // 获取当前客户端已订阅的服务（在本地客户端查询）
            // namingService.getSubscribeServices();
            List<String> serverNameList = namingService.getSubscribeServices().stream().map(ServiceInfo::getName).toList();

            int pageNum = 1;
            final int pageSize = 100;
            String group = config.getRegisterCenter().getNacos().getGroup();

            // 从Nacos服务端获取实例（无论是否订阅）
            List<String> serviceOfServerList = namingService.getServicesOfServer(pageNum, pageSize, group).getData();
            while (CollectionUtil.isNotEmpty(serviceOfServerList)) {
                for (String serverName : serviceOfServerList) {
                    if (!serverNameList.contains(serverName)) { // 没有订阅过
                        NacosRegisterListener listener = new NacosRegisterListener();
                        listener.onEvent(new NamingEvent(serverName, null)); // 首次注册，手动触发一次事件
                        namingService.subscribe(serverName, group, listener); // 当前客户端订阅该服务 （这个项目中就是网关订阅下游服务）
                        log.info("success subscribe group:{} , service:{}", group, serverName);
                    }
                }
                // 下一页
                serviceOfServerList = namingService.getServicesOfServer(++pageNum, pageSize, group).getData();
            }
        } catch (Exception e) {// 捕捉异常防止线程终端
            log.error("出现{}异常", e.getMessage());
        }
    }


    private Properties buildProperty(RegisterCenter registerCenter) {
        Properties properties = new Properties();
        properties.put("serverAddr", registerCenter.getAddress());
        ObjectMapper mapper = new ObjectMapper();
        Map map = mapper.convertValue(registerCenter.getNacos(), Map.class);
        if (map != null && !map.isEmpty()) {
            properties.putAll(map);
        }
        return properties;
    }

    private class NacosRegisterListener implements EventListener {

        @SneakyThrows(Exception.class)
        @Override
        public void onEvent(Event event) {
            // 服务下的实例变化
            if (event instanceof NamingEvent namingEvent) {
                String serviceName = namingEvent.getServiceName();
                String group = config.getRegisterCenter().getNacos().getGroup();

                // 从Nacos拉取服务定义（元数据）信息
                Service service = namingMaintainService.queryService(serviceName, group);
                ServiceDefinition serviceDefinition = new ServiceDefinition(service.getName());
                BeanUtil.fillBeanWithMap(service.getMetadata(), serviceDefinition, true);

                // 从Nacos拉取服务实例信息
                List<Instance> instances = namingService.getAllInstances(serviceName, group);

                // 构建新实例
                Set<ServiceInstance> newInstances = new HashSet<>();
                for (Instance oldInstance : instances) {
                    if (oldInstance == null) continue;
                    ServiceInstance newInstance = new ServiceInstance();
                    BeanUtil.copyProperties(oldInstance, newInstance, true);
                    BeanUtil.fillBeanWithMap(oldInstance.getMetadata(), newInstance, true);
                    newInstances.add(newInstance);
                }
                // 执行自己的监听器（实现延后）
                listener.whenInstanceChange(serviceDefinition, newInstances);
            }
        }
    }
}
